package com.bsoft.server.transport;

import com.bsoft.core.codec.RpcDecoder;
import com.bsoft.core.codec.RpcEncoder;
import com.bsoft.core.protocol.MessageProtocol;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class NettyServer {

    public void start(int port) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        // 当ChannelHandler逻辑复杂，需要处理并发请求时，可以使用netty的EventExecutorGroup机制并发执行ChannelHandler
        EventExecutorGroup executorGroup = new DefaultEventExecutorGroup(16);
        ServerBootstrap bootstrap = new ServerBootstrap();
        try {
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    // 表示系统用于临时存放已完成三次握手的请求的队列的最大长度,如果连接建立频繁，服务器处理创建新连接较慢，可以适当调大这个参数
                    .option(ChannelOption.SO_BACKLOG,128)
                    // 是否开启 TCP 底层心跳机制
                    .childOption(ChannelOption.SO_KEEPALIVE,true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline()
                                    .addLast("rpcDecoder", new RpcDecoder())
                                    .addLast("rpcEncoder", new RpcEncoder<MessageProtocol>())
                                    .addLast(executorGroup, new RpcRequestHandler());
                        }
                    });
            // 同步等待端口绑定成功
            ChannelFuture channelFuture = bootstrap.bind(port).sync();
            log.info("========> rpc server is started for port:{}", port);
            // 同步等待服务器监听通道关闭
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            executorGroup.shutdownGracefully();
        }
    }

}
